/*
 * Copyright 2012-2022 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.springframework.boot.autoconfigure.kafka;

import java.io.IOException;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.DeprecatedConfigurationProperty;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.boot.context.properties.source.MutuallyExclusiveConfigurationPropertiesException;
import org.springframework.boot.convert.DurationUnit;
import org.springframework.core.io.Resource;
import org.springframework.kafka.listener.ContainerProperties.AckMode;
import org.springframework.kafka.security.jaas.KafkaJaasLoginModuleInitializer;
import org.springframework.util.CollectionUtils;
import org.springframework.util.unit.DataSize;

/**
 * Spring for Apache Kafka 的配置属性。
 * <p>
 * 用户应参考 Kafka 文档以获取这些属性的完整描述。
 *
 * @author Gary Russell
 * @author Stephane Nicoll
 * @author Artem Bilan
 * @author Nakul Mishra
 * @author Tomaz Fernandes
 * @since 1.5.0
 */
@ConfigurationProperties(prefix = "spring.kafka")
public class KafkaProperties {

	/**
	 * 用于建立与 Kafka 集群的初始连接的主机:端口对的逗号分隔列表。除非被覆盖，否则此属性适用于所有组件。
	 */
	private List<String> bootstrapServers = new ArrayList<>(Collections.singletonList("localhost:9092"));

	/**
	 * 向服务器发送请求时传递的 ID。用于服务器端日志记录。
	 */
	private String clientId;

	/**
	 * 生产者和消费者共有的、用于配置客户端的额外属性。
	 */
	private final Map<String, String> properties = new HashMap<>();

	private final Consumer consumer = new Consumer();

	private final Producer producer = new Producer();

	private final Admin admin = new Admin();

	private final Streams streams = new Streams();

	private final Listener listener = new Listener();

	private final Ssl ssl = new Ssl();

	private final Jaas jaas = new Jaas();

	private final Template template = new Template();

	private final Security security = new Security();

	private final Retry retry = new Retry();

	public List<String> getBootstrapServers() {
		return this.bootstrapServers;
	}

	public void setBootstrapServers(List<String> bootstrapServers) {
		this.bootstrapServers = bootstrapServers;
	}

	public String getClientId() {
		return this.clientId;
	}

	public void setClientId(String clientId) {
		this.clientId = clientId;
	}

	public Map<String, String> getProperties() {
		return this.properties;
	}

	public Consumer getConsumer() {
		return this.consumer;
	}

	public Producer getProducer() {
		return this.producer;
	}

	public Listener getListener() {
		return this.listener;
	}

	public Admin getAdmin() {
		return this.admin;
	}

	public Streams getStreams() {
		return this.streams;
	}

	public Ssl getSsl() {
		return this.ssl;
	}

	public Jaas getJaas() {
		return this.jaas;
	}

	public Template getTemplate() {
		return this.template;
	}

	public Security getSecurity() {
		return this.security;
	}

	public Retry getRetry() {
		return this.retry;
	}

	private Map<String, Object> buildCommonProperties() {
		Map<String, Object> properties = new HashMap<>();
		if (this.bootstrapServers != null) {
			properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
		}
		if (this.clientId != null) {
			properties.put(CommonClientConfigs.CLIENT_ID_CONFIG, this.clientId);
		}
		properties.putAll(this.ssl.buildProperties());
		properties.putAll(this.security.buildProperties());
		if (!CollectionUtils.isEmpty(this.properties)) {
			properties.putAll(this.properties);
		}
		return properties;
	}

	/**
	 * 根据此实例的状态创建一个初始的消费者属性映射。
	 * <p>
	 * 这允许你在必要时添加额外的属性，并覆盖默认的 kafkaConsumerFactory bean。
	 * @return 使用此实例上定义的自定义设置初始化的消费者属性
	 */
	public Map<String, Object> buildConsumerProperties() {
		Map<String, Object> properties = buildCommonProperties();
		properties.putAll(this.consumer.buildProperties());
		return properties;
	}

	/**
	 * 根据此实例的状态创建一个初始的生产者属性映射。
	 * <p>
	 * 这允许你在必要时添加额外的属性，并覆盖默认的 kafkaProducerFactory bean。
	 * @return 使用此实例上定义的自定义设置初始化的生产者属性
	 */
	public Map<String, Object> buildProducerProperties() {
		Map<String, Object> properties = buildCommonProperties();
		properties.putAll(this.producer.buildProperties());
		return properties;
	}

	/**
	 * 根据此实例的状态创建一个初始的管理员属性映射。
	 * <p>
	 * 这允许你在必要时添加额外的属性，并覆盖默认的 kafkaAdmin bean。
	 * @return 使用此实例上定义的自定义设置初始化的管理员属性
	 */
	public Map<String, Object> buildAdminProperties() {
		Map<String, Object> properties = buildCommonProperties();
		properties.putAll(this.admin.buildProperties());
		return properties;
	}

	/**
	 * 根据此实例的状态创建一个初始的 Streams 属性映射。
	 * <p>
	 * 这允许你在必要时添加额外的属性。
	 * @return 使用此实例上定义的自定义设置初始化的 Streams 属性
	 */
	public Map<String, Object> buildStreamsProperties() {
		Map<String, Object> properties = buildCommonProperties();
		properties.putAll(this.streams.buildProperties());
		return properties;
	}

	public static class Consumer {

		private final Ssl ssl = new Ssl();

		private final Security security = new Security();

		/**
		 * 如果 'enable.auto.commit' 设置为 true，则消费者偏移量自动提交到 Kafka 的频率。
		 */
		private Duration autoCommitInterval;

		/**
		 * 当 Kafka 中没有初始偏移量，或者当前偏移量在服务器上不再存在时，应该如何处理。
		 */
		private String autoOffsetReset;

		/**
		 * 用于建立与 Kafka 集群的初始连接的主机:端口对的逗号分隔列表。此属性会覆盖全局属性，用于消费者。
		 */
		private List<String> bootstrapServers;

		/**
		 * 向服务器发送请求时传递的 ID。用于服务器端日志记录。
		 */
		private String clientId;

		/**
		 * 消费者的偏移量是否在后台定期提交。
		 */
		private Boolean enableAutoCommit;

		/**
		 * 如果没有足够的数据立即满足 "fetch-min-size" 的要求，服务器在响应 fetch 请求之前最大阻塞的时间。
		 */
		private Duration fetchMaxWait;

		/**
		 * 服务器在 fetch 请求中应返回的最小数据量。
		 */
		private DataSize fetchMinSize;

		/**
		 * 唯一标识此消费者所属的消费者组的字符串。
		 */
		private String groupId;

		/**
		 * 发送给消费者协调器的心跳之间的预期时间。
		 */
		private Duration heartbeatInterval;

		/**
		 * 用于读取以事务方式写入的消息的隔离级别。
		 */
		private IsolationLevel isolationLevel = IsolationLevel.READ_UNCOMMITTED;

		/**
		 * 键的解序列化器类。
		 */
		private Class<?> keyDeserializer = StringDeserializer.class;

		/**
		 * 值的解序列化器类。
		 */
		private Class<?> valueDeserializer = StringDeserializer.class;

		/**
		 * 单次调用 poll() 返回的最大记录数。
		 */
		private Integer maxPollRecords;

		/**
		 * 用于配置客户端的、额外的、特定于消费者的属性。
		 */
		private final Map<String, String> properties = new HashMap<>();

		public Ssl getSsl() {
			return this.ssl;
		}

		public Security getSecurity() {
			return this.security;
		}

		public Duration getAutoCommitInterval() {
			return this.autoCommitInterval;
		}

		public void setAutoCommitInterval(Duration autoCommitInterval) {
			this.autoCommitInterval = autoCommitInterval;
		}

		public String getAutoOffsetReset() {
			return this.autoOffsetReset;
		}

		public void setAutoOffsetReset(String autoOffsetReset) {
			this.autoOffsetReset = autoOffsetReset;
		}

		public List<String> getBootstrapServers() {
			return this.bootstrapServers;
		}

		public void setBootstrapServers(List<String> bootstrapServers) {
			this.bootstrapServers = bootstrapServers;
		}

		public String getClientId() {
			return this.clientId;
		}

		public void setClientId(String clientId) {
			this.clientId = clientId;
		}

		public Boolean getEnableAutoCommit() {
			return this.enableAutoCommit;
		}

		public void setEnableAutoCommit(Boolean enableAutoCommit) {
			this.enableAutoCommit = enableAutoCommit;
		}

		public Duration getFetchMaxWait() {
			return this.fetchMaxWait;
		}

		public void setFetchMaxWait(Duration fetchMaxWait) {
			this.fetchMaxWait = fetchMaxWait;
		}

		public DataSize getFetchMinSize() {
			return this.fetchMinSize;
		}

		public void setFetchMinSize(DataSize fetchMinSize) {
			this.fetchMinSize = fetchMinSize;
		}

		public String getGroupId() {
			return this.groupId;
		}

		public void setGroupId(String groupId) {
			this.groupId = groupId;
		}

		public Duration getHeartbeatInterval() {
			return this.heartbeatInterval;
		}

		public void setHeartbeatInterval(Duration heartbeatInterval) {
			this.heartbeatInterval = heartbeatInterval;
		}

		public IsolationLevel getIsolationLevel() {
			return this.isolationLevel;
		}

		public void setIsolationLevel(IsolationLevel isolationLevel) {
			this.isolationLevel = isolationLevel;
		}

		public Class<?> getKeyDeserializer() {
			return this.keyDeserializer;
		}

		public void setKeyDeserializer(Class<?> keyDeserializer) {
			this.keyDeserializer = keyDeserializer;
		}

		public Class<?> getValueDeserializer() {
			return this.valueDeserializer;
		}

		public void setValueDeserializer(Class<?> valueDeserializer) {
			this.valueDeserializer = valueDeserializer;
		}

		public Integer getMaxPollRecords() {
			return this.maxPollRecords;
		}

		public void setMaxPollRecords(Integer maxPollRecords) {
			this.maxPollRecords = maxPollRecords;
		}

		public Map<String, String> getProperties() {
			return this.properties;
		}

		public Map<String, Object> buildProperties() {
			Properties properties = new Properties();
			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
			map.from(this::getAutoCommitInterval).asInt(Duration::toMillis)
					.to(properties.in(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG));
			map.from(this::getAutoOffsetReset).to(properties.in(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG));
			map.from(this::getBootstrapServers).to(properties.in(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG));
			map.from(this::getClientId).to(properties.in(ConsumerConfig.CLIENT_ID_CONFIG));
			map.from(this::getEnableAutoCommit).to(properties.in(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG));
			map.from(this::getFetchMaxWait).asInt(Duration::toMillis)
					.to(properties.in(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG));
			map.from(this::getFetchMinSize).asInt(DataSize::toBytes)
					.to(properties.in(ConsumerConfig.FETCH_MIN_BYTES_CONFIG));
			map.from(this::getGroupId).to(properties.in(ConsumerConfig.GROUP_ID_CONFIG));
			map.from(this::getHeartbeatInterval).asInt(Duration::toMillis)
					.to(properties.in(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG));
			map.from(() -> getIsolationLevel().name().toLowerCase(Locale.ROOT))
					.to(properties.in(ConsumerConfig.ISOLATION_LEVEL_CONFIG));
			map.from(this::getKeyDeserializer).to(properties.in(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG));
			map.from(this::getValueDeserializer).to(properties.in(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG));
			map.from(this::getMaxPollRecords).to(properties.in(ConsumerConfig.MAX_POLL_RECORDS_CONFIG));
			return properties.with(this.ssl, this.security, this.properties);
		}

	}

	public static class Producer {

		private final Ssl ssl = new Ssl();

		private final Security security = new Security();

		/**
		 * 在认为请求完成之前，生产者需要领导者收到的确认数。
		 */
		private String acks;

		/**
		 * 默认的批处理大小。较小的批处理大小会使批处理不那么常见，并可能降低吞吐量（批处理大小为零则完全禁用批处理）。
		 */
		private DataSize batchSize;

		/**
		 * 用于建立与 Kafka 集群的初始连接的主机:端口对的逗号分隔列表。此属性会覆盖全局属性，用于生产者。
		 */
		private List<String> bootstrapServers;

		/**
		 * 生产者可用于缓冲等待发送到服务器的记录的总内存大小。
		 */
		private DataSize bufferMemory;

		/**
		 * 向服务器发送请求时传递的 ID。用于服务器端日志记录。
		 */
		private String clientId;

		/**
		 * 生产者生成的所有数据的压缩类型。
		 */
		private String compressionType;

		/**
		 * 键的序列化器类。
		 */
		private Class<?> keySerializer = StringSerializer.class;

		/**
		 * 值的序列化器类。
		 */
		private Class<?> valueSerializer = StringSerializer.class;

		/**
		 * 当大于零时，启用对失败发送的重试。
		 */
		private Integer retries;

		/**
		 * 当非空时，为生产者启用事务支持。
		 */
		private String transactionIdPrefix;

		/**
		 * 用于配置客户端的、额外的、特定于生产者的属性。
		 */
		private final Map<String, String> properties = new HashMap<>();

		public Ssl getSsl() {
			return this.ssl;
		}

		public Security getSecurity() {
			return this.security;
		}

		public String getAcks() {
			return this.acks;
		}

		public void setAcks(String acks) {
			this.acks = acks;
		}

		public DataSize getBatchSize() {
			return this.batchSize;
		}

		public void setBatchSize(DataSize batchSize) {
			this.batchSize = batchSize;
		}

		public List<String> getBootstrapServers() {
			return this.bootstrapServers;
		}

		public void setBootstrapServers(List<String> bootstrapServers) {
			this.bootstrapServers = bootstrapServers;
		}

		public DataSize getBufferMemory() {
			return this.bufferMemory;
		}

		public void setBufferMemory(DataSize bufferMemory) {
			this.bufferMemory = bufferMemory;
		}

		public String getClientId() {
			return this.clientId;
		}

		public void setClientId(String clientId) {
			this.clientId = clientId;
		}

		public String getCompressionType() {
			return this.compressionType;
		}

		public void setCompressionType(String compressionType) {
			this.compressionType = compressionType;
		}

		public Class<?> getKeySerializer() {
			return this.keySerializer;
		}

		public void setKeySerializer(Class<?> keySerializer) {
			this.keySerializer = keySerializer;
		}

		public Class<?> getValueSerializer() {
			return this.valueSerializer;
		}

		public void setValueSerializer(Class<?> valueSerializer) {
			this.valueSerializer = valueSerializer;
		}

		public Integer getRetries() {
			return this.retries;
		}

		public void setRetries(Integer retries) {
			this.retries = retries;
		}

		public String getTransactionIdPrefix() {
			return this.transactionIdPrefix;
		}

		public void setTransactionIdPrefix(String transactionIdPrefix) {
			this.transactionIdPrefix = transactionIdPrefix;
		}

		public Map<String, String> getProperties() {
			return this.properties;
		}

		public Map<String, Object> buildProperties() {
			Properties properties = new Properties();
			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
			map.from(this::getAcks).to(properties.in(ProducerConfig.ACKS_CONFIG));
			map.from(this::getBatchSize).asInt(DataSize::toBytes).to(properties.in(ProducerConfig.BATCH_SIZE_CONFIG));
			map.from(this::getBootstrapServers).to(properties.in(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG));
			map.from(this::getBufferMemory).as(DataSize::toBytes)
					.to(properties.in(ProducerConfig.BUFFER_MEMORY_CONFIG));
			map.from(this::getClientId).to(properties.in(ProducerConfig.CLIENT_ID_CONFIG));
			map.from(this::getCompressionType).to(properties.in(ProducerConfig.COMPRESSION_TYPE_CONFIG));
			map.from(this::getKeySerializer).to(properties.in(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG));
			map.from(this::getRetries).to(properties.in(ProducerConfig.RETRIES_CONFIG));
			map.from(this::getValueSerializer).to(properties.in(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG));
			return properties.with(this.ssl, this.security, this.properties);
		}

	}

	public static class Admin {

		private final Ssl ssl = new Ssl();

		private final Security security = new Security();

		/**
		 * 向服务器发送请求时传递的 ID。用于服务器端日志记录。
		 */
		private String clientId;

		/**
		 * 用于配置客户端的、额外的、特定于管理员的属性。
		 */
		private final Map<String, String> properties = new HashMap<>();

		/**
		 * 如果代理在启动时不可用，是否快速失败。
		 */
		private boolean failFast;

		public Ssl getSsl() {
			return this.ssl;
		}

		public Security getSecurity() {
			return this.security;
		}

		public String getClientId() {
			return this.clientId;
		}

		public void setClientId(String clientId) {
			this.clientId = clientId;
		}

		public boolean isFailFast() {
			return this.failFast;
		}

		public void setFailFast(boolean failFast) {
			this.failFast = failFast;
		}

		public Map<String, String> getProperties() {
			return this.properties;
		}

		public Map<String, Object> buildProperties() {
			Properties properties = new Properties();
			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
			map.from(this::getClientId).to(properties.in(ProducerConfig.CLIENT_ID_CONFIG));
			return properties.with(this.ssl, this.security, this.properties);
		}

	}

	/**
	 * 高（和一些中等）优先级的 Streams 属性和一个通用属性桶。
	 */
	public static class Streams {

		private final Ssl ssl = new Ssl();

		private final Security security = new Security();

		private final Cleanup cleanup = new Cleanup();

		/**
		 * Kafka Streams 应用程序的 application.id 属性；默认为 spring.application.name。
		 */
		private String applicationId;

		/**
		 * 是否自动启动 Streams 工厂 bean。
		 */
		private boolean autoStartup = true;

		/**
		 * 用于建立与 Kafka 集群的初始连接的主机:端口对的逗号分隔列表。此属性会覆盖全局属性。
		 */
		private List<String> bootstrapServers;

		/**
		 * 用于跨所有线程进行缓冲的最大内存大小。
		 */
		private DataSize cacheMaxSizeBuffering;

		/**
		 * 向服务器发送请求时传递的 ID。用于服务器端日志记录。
		 */
		private String clientId;

		/**
		 * 流处理应用程序创建的变更日志主题和重新分区主题的复制因子。
		 */
		private Integer replicationFactor;

		/**
		 * 状态存储的目录位置。
		 */
		private String stateDir;

		/**
		 * 用于配置 Streams 的附加 Kafka 属性。
		 */
		private final Map<String, String> properties = new HashMap<>();

		public Ssl getSsl() {
			return this.ssl;
		}

		public Security getSecurity() {
			return this.security;
		}

		public Cleanup getCleanup() {
			return this.cleanup;
		}

		public String getApplicationId() {
			return this.applicationId;
		}

		public void setApplicationId(String applicationId) {
			this.applicationId = applicationId;
		}

		public boolean isAutoStartup() {
			return this.autoStartup;
		}

		public void setAutoStartup(boolean autoStartup) {
			this.autoStartup = autoStartup;
		}

		public List<String> getBootstrapServers() {
			return this.bootstrapServers;
		}

		public void setBootstrapServers(List<String> bootstrapServers) {
			this.bootstrapServers = bootstrapServers;
		}

		public DataSize getCacheMaxSizeBuffering() {
			return this.cacheMaxSizeBuffering;
		}

		public void setCacheMaxSizeBuffering(DataSize cacheMaxSizeBuffering) {
			this.cacheMaxSizeBuffering = cacheMaxSizeBuffering;
		}

		public String getClientId() {
			return this.clientId;
		}

		public void setClientId(String clientId) {
			this.clientId = clientId;
		}

		public Integer getReplicationFactor() {
			return this.replicationFactor;
		}

		public void setReplicationFactor(Integer replicationFactor) {
			this.replicationFactor = replicationFactor;
		}

		public String getStateDir() {
			return this.stateDir;
		}

		public void setStateDir(String stateDir) {
			this.stateDir = stateDir;
		}

		public Map<String, String> getProperties() {
			return this.properties;
		}

		public Map<String, Object> buildProperties() {
			Properties properties = new Properties();
			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
			map.from(this::getApplicationId).to(properties.in("application.id"));
			map.from(this::getBootstrapServers).to(properties.in(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG));
			map.from(this::getCacheMaxSizeBuffering).asInt(DataSize::toBytes)
					.to(properties.in("cache.max.bytes.buffering"));
			map.from(this::getClientId).to(properties.in(CommonClientConfigs.CLIENT_ID_CONFIG));
			map.from(this::getReplicationFactor).to(properties.in("replication.factor"));
			map.from(this::getStateDir).to(properties.in("state.dir"));
			return properties.with(this.ssl, this.security, this.properties);
		}

	}

	public static class Template {

		/**
		 * 发送消息时使用的默认主题。
		 */
		private String defaultTopic;

		/**
		 * 事务ID前缀，将覆盖生产者工厂中的事务ID前缀。
		 */
		private String transactionIdPrefix;

		public String getDefaultTopic() {
			return this.defaultTopic;
		}

		public void setDefaultTopic(String defaultTopic) {
			this.defaultTopic = defaultTopic;
		}

		public String getTransactionIdPrefix() {
			return this.transactionIdPrefix;
		}

		public void setTransactionIdPrefix(String transactionIdPrefix) {
			this.transactionIdPrefix = transactionIdPrefix;
		}

	}

	public static class Listener {

		public enum Type {

			/**
			 * 每次用一个 ConsumerRecord 调用端点。
			 */
			SINGLE,

			/**
			 * 每次用一批 ConsumerRecord 调用端点。
			 */
			BATCH

		}

		/**
		 * 监听器类型。
		 */
		private Type type = Type.SINGLE;

		/**
		 * 监听器的确认模式（AckMode）。请参阅 spring-kafka 文档。
		 */
		private AckMode ackMode;

		/**
		 * 监听器消费者的 client.id 属性的前缀。
		 */
		private String clientId;

		/**
		 * 在监听器容器中运行的线程数。
		 */
		private Integer concurrency;

		/**
		 * 轮询消费者时使用的超时时间。
		 */
		private Duration pollTimeout;

		/**
		 * 应用于 "pollTimeout" 的乘数，用于确定消费者是否无响应。
		 */
		private Float noPollThreshold;

		/**
		 * 当确认模式（ackMode）为 "COUNT" 或 "COUNT_TIME" 时，两次偏移量提交之间的记录数。
		 */
		private Integer ackCount;

		/**
		 * 当确认模式（ackMode）为 "TIME" 或 "COUNT_TIME" 时，两次偏移量提交之间的时间。
		 */
		private Duration ackTime;

		/**
		 * 消费者 {@code Consumer.poll(Duration)} 调用之间的休眠间隔。
		 */
		private Duration idleBetweenPolls = Duration.ZERO;

		/**
		 * 发布空闲消费者事件（未收到数据）之间的时间。
		 */
		private Duration idleEventInterval;

		/**
		 * 发布空闲分区消费者事件（分区未收到数据）之间的时间。
		 */
		private Duration idlePartitionEventInterval;

		/**
		 * 检查无响应消费者之间的时间。如果未指定持续时间后缀，则使用秒作为单位。
		 */
		@DurationUnit(ChronoUnit.SECONDS)
		private Duration monitorInterval;

		/**
		 * 在初始化期间是否记录容器配置（INFO 级别）。
		 */
		private Boolean logContainerConfig;

		/**
		 * 当尝试重试时，是否禁止将整个记录写入日志。
		 */
		private boolean onlyLogRecordMetadata = true;

		/**
		 * 如果代理上至少有一个配置的主题不存在，容器是否应停止启动。
		 */
		private boolean missingTopicsFatal = false;

		/**
		 * 容器是在处理完当前记录后停止，还是在处理完上次拉取的所有记录后停止。
		 */
		private boolean immediateStop = false;

		public Type getType() {
			return this.type;
		}

		public void setType(Type type) {
			this.type = type;
		}

		public AckMode getAckMode() {
			return this.ackMode;
		}

		public void setAckMode(AckMode ackMode) {
			this.ackMode = ackMode;
		}

		public String getClientId() {
			return this.clientId;
		}

		public void setClientId(String clientId) {
			this.clientId = clientId;
		}

		public Integer getConcurrency() {
			return this.concurrency;
		}

		public void setConcurrency(Integer concurrency) {
			this.concurrency = concurrency;
		}

		public Duration getPollTimeout() {
			return this.pollTimeout;
		}

		public void setPollTimeout(Duration pollTimeout) {
			this.pollTimeout = pollTimeout;
		}

		public Float getNoPollThreshold() {
			return this.noPollThreshold;
		}

		public void setNoPollThreshold(Float noPollThreshold) {
			this.noPollThreshold = noPollThreshold;
		}

		public Integer getAckCount() {
			return this.ackCount;
		}

		public void setAckCount(Integer ackCount) {
			this.ackCount = ackCount;
		}

		public Duration getAckTime() {
			return this.ackTime;
		}

		public void setAckTime(Duration ackTime) {
			this.ackTime = ackTime;
		}

		public Duration getIdleBetweenPolls() {
			return this.idleBetweenPolls;
		}

		public void setIdleBetweenPolls(Duration idleBetweenPolls) {
			this.idleBetweenPolls = idleBetweenPolls;
		}

		public Duration getIdleEventInterval() {
			return this.idleEventInterval;
		}

		public void setIdleEventInterval(Duration idleEventInterval) {
			this.idleEventInterval = idleEventInterval;
		}

		public Duration getIdlePartitionEventInterval() {
			return this.idlePartitionEventInterval;
		}

		public void setIdlePartitionEventInterval(Duration idlePartitionEventInterval) {
			this.idlePartitionEventInterval = idlePartitionEventInterval;
		}

		public Duration getMonitorInterval() {
			return this.monitorInterval;
		}

		public void setMonitorInterval(Duration monitorInterval) {
			this.monitorInterval = monitorInterval;
		}

		public Boolean getLogContainerConfig() {
			return this.logContainerConfig;
		}

		public void setLogContainerConfig(Boolean logContainerConfig) {
			this.logContainerConfig = logContainerConfig;
		}

		@Deprecated
		@DeprecatedConfigurationProperty(reason = "Use KafkaUtils#setConsumerRecordFormatter instead.")
		public boolean isOnlyLogRecordMetadata() {
			return this.onlyLogRecordMetadata;
		}

		@Deprecated
		public void setOnlyLogRecordMetadata(boolean onlyLogRecordMetadata) {
			this.onlyLogRecordMetadata = onlyLogRecordMetadata;
		}

		public boolean isMissingTopicsFatal() {
			return this.missingTopicsFatal;
		}

		public void setMissingTopicsFatal(boolean missingTopicsFatal) {
			this.missingTopicsFatal = missingTopicsFatal;
		}

		public boolean isImmediateStop() {
			return this.immediateStop;
		}

		public void setImmediateStop(boolean immediateStop) {
			this.immediateStop = immediateStop;
		}

	}

	public static class Ssl {

		/**
		 * 密钥库密钥或密钥库文件中的私钥密码。
		 */
		private String keyPassword;

		/**
		 * PEM 格式的证书链，包含一个 X.509 证书列表。
		 */
		private String keyStoreCertificateChain;

		/**
		 * PEM 格式的私钥，采用 PKCS#8 密钥。
		 */
		private String keyStoreKey;

		/**
		 * 密钥库文件的位置。
		 */
		private Resource keyStoreLocation;

		/**
		 * 密钥库文件的存储密码。
		 */
		private String keyStorePassword;

		/**
		 * 密钥库的类型。
		 */
		private String keyStoreType;

		/**
		 * PEM 格式的受信任证书，包含 X.509 证书。
		 */
		private String trustStoreCertificates;

		/**
		 * 信任库文件的位置。
		 */
		private Resource trustStoreLocation;

		/**
		 * 信任库文件的存储密码。
		 */
		private String trustStorePassword;

		/**
		 * 信任库的类型。
		 */
		private String trustStoreType;

		/**
		 * 要使用的 SSL 协议。
		 */
		private String protocol;

		public String getKeyPassword() {
			return this.keyPassword;
		}

		public void setKeyPassword(String keyPassword) {
			this.keyPassword = keyPassword;
		}

		public String getKeyStoreCertificateChain() {
			return this.keyStoreCertificateChain;
		}

		public void setKeyStoreCertificateChain(String keyStoreCertificateChain) {
			this.keyStoreCertificateChain = keyStoreCertificateChain;
		}

		public String getKeyStoreKey() {
			return this.keyStoreKey;
		}

		public void setKeyStoreKey(String keyStoreKey) {
			this.keyStoreKey = keyStoreKey;
		}

		public Resource getKeyStoreLocation() {
			return this.keyStoreLocation;
		}

		public void setKeyStoreLocation(Resource keyStoreLocation) {
			this.keyStoreLocation = keyStoreLocation;
		}

		public String getKeyStorePassword() {
			return this.keyStorePassword;
		}

		public void setKeyStorePassword(String keyStorePassword) {
			this.keyStorePassword = keyStorePassword;
		}

		public String getKeyStoreType() {
			return this.keyStoreType;
		}

		public void setKeyStoreType(String keyStoreType) {
			this.keyStoreType = keyStoreType;
		}

		public String getTrustStoreCertificates() {
			return this.trustStoreCertificates;
		}

		public void setTrustStoreCertificates(String trustStoreCertificates) {
			this.trustStoreCertificates = trustStoreCertificates;
		}

		public Resource getTrustStoreLocation() {
			return this.trustStoreLocation;
		}

		public void setTrustStoreLocation(Resource trustStoreLocation) {
			this.trustStoreLocation = trustStoreLocation;
		}

		public String getTrustStorePassword() {
			return this.trustStorePassword;
		}

		public void setTrustStorePassword(String trustStorePassword) {
			this.trustStorePassword = trustStorePassword;
		}

		public String getTrustStoreType() {
			return this.trustStoreType;
		}

		public void setTrustStoreType(String trustStoreType) {
			this.trustStoreType = trustStoreType;
		}

		public String getProtocol() {
			return this.protocol;
		}

		public void setProtocol(String protocol) {
			this.protocol = protocol;
		}

		public Map<String, Object> buildProperties() {
			validate();
			Properties properties = new Properties();
			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
			map.from(this::getKeyPassword).to(properties.in(SslConfigs.SSL_KEY_PASSWORD_CONFIG));
			map.from(this::getKeyStoreCertificateChain)
					.to(properties.in(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG));
			map.from(this::getKeyStoreKey).to(properties.in(SslConfigs.SSL_KEYSTORE_KEY_CONFIG));
			map.from(this::getKeyStoreLocation).as(this::resourceToPath)
					.to(properties.in(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG));
			map.from(this::getKeyStorePassword).to(properties.in(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG));
			map.from(this::getKeyStoreType).to(properties.in(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG));
			map.from(this::getTrustStoreCertificates).to(properties.in(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG));
			map.from(this::getTrustStoreLocation).as(this::resourceToPath)
					.to(properties.in(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG));
			map.from(this::getTrustStorePassword).to(properties.in(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG));
			map.from(this::getTrustStoreType).to(properties.in(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG));
			map.from(this::getProtocol).to(properties.in(SslConfigs.SSL_PROTOCOL_CONFIG));
			return properties;
		}

		private void validate() {
			MutuallyExclusiveConfigurationPropertiesException.throwIfMultipleNonNullValuesIn((entries) -> {
				entries.put("spring.kafka.ssl.key-store-key", this.getKeyStoreKey());
				entries.put("spring.kafka.ssl.key-store-location", this.getKeyStoreLocation());
			});
			MutuallyExclusiveConfigurationPropertiesException.throwIfMultipleNonNullValuesIn((entries) -> {
				entries.put("spring.kafka.ssl.trust-store-certificates", this.getTrustStoreCertificates());
				entries.put("spring.kafka.ssl.trust-store-location", this.getTrustStoreLocation());
			});
		}

		private String resourceToPath(Resource resource) {
			try {
				return resource.getFile().getAbsolutePath();
			}
			catch (IOException ex) {
				throw new IllegalStateException("Resource '" + resource + "' must be on a file system", ex);
			}
		}

	}

	public static class Jaas {

		/**
		 * 是否启用JAAS配置。
		 */
		private boolean enabled;

		/**
		 * 登录模块。
		 */
		private String loginModule = "com.sun.security.auth.module.Krb5LoginModule";

		/**
		 * 登录配置的控制标志。
		 */
		private KafkaJaasLoginModuleInitializer.ControlFlag controlFlag = KafkaJaasLoginModuleInitializer.ControlFlag.REQUIRED;

		/**
		 * 附加的JAAS选项。
		 */
		private final Map<String, String> options = new HashMap<>();

		public boolean isEnabled() {
			return this.enabled;
		}

		public void setEnabled(boolean enabled) {
			this.enabled = enabled;
		}

		public String getLoginModule() {
			return this.loginModule;
		}

		public void setLoginModule(String loginModule) {
			this.loginModule = loginModule;
		}

		public KafkaJaasLoginModuleInitializer.ControlFlag getControlFlag() {
			return this.controlFlag;
		}

		public void setControlFlag(KafkaJaasLoginModuleInitializer.ControlFlag controlFlag) {
			this.controlFlag = controlFlag;
		}

		public Map<String, String> getOptions() {
			return this.options;
		}

		public void setOptions(Map<String, String> options) {
			if (options != null) {
				this.options.putAll(options);
			}
		}

	}

	public static class Security {

		/**
		 * 用于与代理通信的安全协议。
		 */
		private String protocol;

		public String getProtocol() {
			return this.protocol;
		}

		public void setProtocol(String protocol) {
			this.protocol = protocol;
		}

		public Map<String, Object> buildProperties() {
			Properties properties = new Properties();
			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
			map.from(this::getProtocol).to(properties.in(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
			return properties;
		}

	}

	public static class Retry {

		private final Topic topic = new Topic();

		public Topic getTopic() {
			return this.topic;
		}

		/**
		 * 用于非阻塞、基于主题的重试属性。
		 */
		public static class Topic {

			/**
			 * 是否启用基于主题的非阻塞重试。
			 */
			private boolean enabled;

			/**
			 * 在将消息发送到 DLT（死信主题）之前，总共尝试处理的次数。
			 */
			private int attempts = 3;

			/**
			 * 规范的回退周期。在指数回退的情况下用作初始值，在统一回退的情况下用作最小值。
			 */
			private Duration delay = Duration.ofSeconds(1);

			/**
			 * 用于生成下一个回退延迟的乘数。
			 */
			private double multiplier = 0.0;

			/**
			 * 重试之间的最大等待时间。如果小于延迟时间，则应用默认的 30 秒。
			 */
			private Duration maxDelay = Duration.ZERO;

			/**
			 * 是否具有回退延迟。
			 */
			private boolean randomBackOff = false;

			public boolean isEnabled() {
				return this.enabled;
			}

			public void setEnabled(boolean enabled) {
				this.enabled = enabled;
			}

			public int getAttempts() {
				return this.attempts;
			}

			public void setAttempts(int attempts) {
				this.attempts = attempts;
			}

			public Duration getDelay() {
				return this.delay;
			}

			public void setDelay(Duration delay) {
				this.delay = delay;
			}

			public double getMultiplier() {
				return this.multiplier;
			}

			public void setMultiplier(double multiplier) {
				this.multiplier = multiplier;
			}

			public Duration getMaxDelay() {
				return this.maxDelay;
			}

			public void setMaxDelay(Duration maxDelay) {
				this.maxDelay = maxDelay;
			}

			public boolean isRandomBackOff() {
				return this.randomBackOff;
			}

			public void setRandomBackOff(boolean randomBackOff) {
				this.randomBackOff = randomBackOff;
			}

		}

	}

	public static class Cleanup {

		/**
		 * 在启动时清理应用程序的本地状态目录。
		 */
		private boolean onStartup = false;

		/**
		 * 在关闭时清理应用程序的本地状态目录。
		 */
		private boolean onShutdown = false;

		public boolean isOnStartup() {
			return this.onStartup;
		}

		public void setOnStartup(boolean onStartup) {
			this.onStartup = onStartup;
		}

		public boolean isOnShutdown() {
			return this.onShutdown;
		}

		public void setOnShutdown(boolean onShutdown) {
			this.onShutdown = onShutdown;
		}

	}

	public enum IsolationLevel {

		/**
		 * 读取所有内容，包括已中止的事务。
		 */
		READ_UNCOMMITTED((byte) 0),

		/**
		 * 读取来自已提交事务的记录，以及不属于任何事务的记录。
		 */
		READ_COMMITTED((byte) 1);

		private final byte id;

		IsolationLevel(byte id) {
			this.id = id;
		}

		public byte id() {
			return this.id;
		}

	}

	@SuppressWarnings("serial")
	private static class Properties extends HashMap<String, Object> {

		<V> java.util.function.Consumer<V> in(String key) {
			return (value) -> put(key, value);
		}

		Properties with(Ssl ssl, Security security, Map<String, String> properties) {
			putAll(ssl.buildProperties());
			putAll(security.buildProperties());
			putAll(properties);
			return this;
		}

	}

}
